package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/8/23 14:18
 */
public class FieldExtensionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema = Schema.newBuilder()
                .column("id", DataTypes.BIGINT())
                .column("sid", DataTypes.STRING())
                .column("ts", DataTypes.TIMESTAMP(3))
                .column("flag", DataTypes.BOOLEAN())
                .column("vc", DataTypes.INT())
                .watermark("ts", "ts - INTERVAL '3' SECOND")
                .build();
        Table table = tableEnv.from(
                TableDescriptor.forConnector("datagen")
                        .schema(schema)
                        .option("rows-per-second", "1")
//                        .option("number-of-rows", "10")
                        .option("fields.sid.length", "5")
                        .option("fields.id.min", "0")
                        .option("fields.id.max", "10")
                        .option("fields.vc.min", "50")
                        .option("fields.vc.max", "100")
                        .build()
        );
        tableEnv.createTemporaryView("source_table", table);

        tableEnv.executeSql("CREATE TEMPORARY VIEW extension_table as " +
                "select source_table.`id`," +
                "source_table.`sid`," +
                "source_table.`ts`," +
                "source_table.`flag`," +
                "source_table.`vc`," +
                "CAST('告警事件名称' as STRING) as ext_field1 ," +
                "CAST('12' as INT) as ext_field2 " +
                "from source_table");


//        Table sinkTable = tableEnv.sqlQuery("select * from extension_table");
//        tableEnv.toDataStream(sinkTable).print();
//        env.execute();

        tableEnv.executeSql("create table sink_table(" +
                "id BIGINT," +
                "sid STRING," +
                "ts TIMESTAMP(3)," +
                "flag BOOLEAN," +
                "vc INT," +
                "ext_field1 STRING," +
                "ext_field2 INT" +
                ") with ('connector' = 'print')");

        tableEnv.executeSql("insert into sink_table select id,sid,ts,flag,vc, ext_field1, ext_field2 from extension_table");

    }
}
